/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package aiaudio.lastfm.hbase;

import aiaudio.database.tables.SocialTable;
import aiaudio.database.tables.RatingsListTable;
import aiaudio.database.tables.DataMetricsTable;
import aiaudio.database.tables.UsersListTable;
import aiaudio.database.tables.ArtistsListTable;
import de.umass.lastfm.Artist;
import de.umass.lastfm.User;
import java.io.IOException;
import java.util.Calendar;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import aiaudio.Application;
import aiaudio.database.tablegroup.InitialDataTableGroup;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;

/**
 *
 * @author nastya
 */
public class SchemeCommunicator {

    private HTable ratingsTable;
    private HTable usersTable;
    private HTable tracksTable;
    private HTable dataMetricsTable;
    private final HTable socialLinkTable;

    public SchemeCommunicator() throws IOException, CannotCreateTableException {
        InitialDataTableGroup.getInstance().create(Application.database());
        
        usersTable = Application.database().getByName(InitialDataTableGroup.getInstance().getUsersTable()).getTable();
        ratingsTable = Application.database().getByName(InitialDataTableGroup.getInstance().getRatingListTable()).getTable();
        tracksTable = Application.database().getByName(InitialDataTableGroup.getInstance().getArtistsTable()).getTable();
        dataMetricsTable = Application.database().getByName(InitialDataTableGroup.getInstance().getDataMetricsTable()).getTable();
        socialLinkTable = Application.database().getByName(InitialDataTableGroup.getInstance().getSocialLinkTable()).getTable();
    }

    public HTable getUsersTable() {
        return usersTable;
    }

    public void insertUser(User u) throws IOException {
        if (u == null) {
            System.out.println("user is null");
            return;
        }

        if (usersTable == null) {
            System.out.println("Unable to insert user: no users table");
            return;
        }


        String rowKey = u.getId();
        if (rowKey == null) {
            System.out.println("add " + u.getName() + " to database failed: id is null");
            return;
        }
        final byte[] rkb = Bytes.toBytes(rowKey);

        Get get = new Get(rkb);
        boolean exists = usersTable.exists(get);
        if (exists) {
            System.out.println("User " + u.getId() + " (" + u.getName() + ") exists in table");
            return;
        }
        Put put = new Put(rkb);
        UsersListTable.addAge(put, u.getAge());
        UsersListTable.addPlaycount(put, u.getPlaycount());
        UsersListTable.addCountry(put, u.getCountry());
        UsersListTable.addGender(put, u.getGender());
        UsersListTable.addRealname(put, u.getRealname());
        UsersListTable.addURL(put, u.getUrl());
        UsersListTable.addUsername(put, u.getName());
        usersTable.put(put);
        DataMetricsTable.incrementUsersCount(dataMetricsTable);
        System.out.println("added user: " + rowKey);
    }

    public void createLink(String id1, String id2) throws IOException {
        if (id1 == null || id1.isEmpty()) {
            return;
        }

        if (id2 == null || id2.isEmpty()) {
            return;
        }

        System.out.println("create link between " + id1 + " and " + id2);
        String key = Calendar.getInstance().getTimeInMillis() + "";

        Put put = new Put(Bytes.toBytes(key));
        SocialTable.addLink(put, id1, id2);
        socialLinkTable.put(put);
    }

    public Filter createRatingByArtistAnsUserFilter(String userKey1, String userKey2) {
        Filter ratingByUserFilter1 = createRatingByUserFilter(userKey1);
        Filter ratingByUserFilter2 = createRatingByUserFilter(userKey2);

        FilterList list = new FilterList();
        list.addFilter(ratingByUserFilter1);
        list.addFilter(ratingByUserFilter2);

        return list;
    }

    public Filter createRatingByUserFilter(String userKey) {
        return new SingleColumnValueFilter(Bytes.toBytes("s"), Bytes.toBytes("f1"), CompareOp.EQUAL, Bytes.toBytes(userKey));
    }

    public String insertArtist(Artist artist) throws IOException {
        if (tracksTable == null) {
            System.out.println("Unable to insert artist: no artists table");
            return "";
        }


        String rowKey;
        rowKey = createArtistKey(artist);

        if (rowKey == null || rowKey.isEmpty()) {
            return "";
        }

        final byte[] rkb = Bytes.toBytes(rowKey);
        Get get = new Get(rkb);
        boolean exists = tracksTable.exists(get);
        if (exists) {
            //System.out.println("Artist " + rowKey + " (" + artist.getName() + ") exists in table");
            return rowKey;
        }

        Put put = new Put(rkb);
        ArtistsListTable.addMbid(put, artist.getMbid());
        ArtistsListTable.addName(put, artist.getName());
        ArtistsListTable.addUrl(put, artist.getUrl());
        ArtistsListTable.addFirstVoting(put);
        tracksTable.put(put);
        DataMetricsTable.incrementArtistsCount(dataMetricsTable);
        return rowKey;
    }

    protected String createArtistKey(Artist artist) {
        String rowKey;
        if (artist.getMbid() == null || artist.getMbid().trim().isEmpty()) {
            rowKey = artist.getName();
        } else {
            rowKey = artist.getMbid();
        }
        return rowKey;
    }

    public void insertRating(String userId, String artistId, int playcount) throws IOException {
        if (ratingsTable == null) {
            System.out.println("Unable to insert rating: no ratings table");
        }

        String rowKey = Calendar.getInstance().getTime().getTime() + "";
        final byte[] rkb = Bytes.toBytes(rowKey);

        Put put = new Put(rkb);
        RatingsListTable.addArtist(put, artistId);
        RatingsListTable.addUser(put, userId);
        RatingsListTable.addListenCount(put, playcount);

        DataMetricsTable.incrementRatingsCount(dataMetricsTable);
        ArtistsListTable.incrementVoting(ratingsTable, artistId);
        ArtistsListTable.incrementListenCount(ratingsTable, artistId, (long) playcount);
        UsersListTable.incrementVoting(usersTable, userId);
        UsersListTable.incrementListenCount(usersTable, userId, playcount);
        ratingsTable.put(put);

    }
}
